package com.openness.spider.commons;

import java.io.IOException;
import java.util.Collection;
import java.util.HashSet;
import java.util.Set;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.openness.spider.SpiderConfig;

/**
 * @author Rain
 * @version 2013-6-20 下午10:01:30
 */

public abstract class GeneralBuffer<T> extends Configurable implements Runnable {

	private static final Logger LOGGER = LoggerFactory
			.getLogger(GeneralBuffer.class);

	private Set<T> buffer = new HashSet<T>();

	private long lastFlushTime = System.currentTimeMillis();

	private boolean stop = false;

	public GeneralBuffer(SpiderConfig config) {
		super(config);
	}

	public void put(T t) throws InterruptedException {
		if (t == null) {
			return;
		}

		synchronized (buffer) {
			while (!stop && t != null) {
				if (buffer.size() < getBufferSize()) {
					buffer.add(t);

					t = null;
				} else {
					buffer.wait();
				}
			}
		}
	}

	public void put(Collection<T> ts) throws InterruptedException {
		if (ts == null || ts.size() == 0) {
			return;
		}

		synchronized (buffer) {
			while (!stop && ts != null) {
				if (buffer.size() <= getBufferSize()) {
					buffer.addAll(ts);

					ts = null;
				} else {
					buffer.wait();
				}
			}
		}
	}

	public abstract void onStart() throws Exception;

	public abstract void onStop() throws Exception;

	public abstract void flush(Set<T> buffer) throws IOException;

	public abstract long getBufferSize();

	public abstract long getFlushInternal();

	public abstract long getSleepTime();

	@Override
	public void run() {
		try {
			onStart();
		} catch (Exception e) {
			throw new RuntimeException(Statics.HOSTNAME + "\t"
					+ Thread.currentThread().getName() + "\t" + e.toString());
		}

		while (!stop) {
			long now = System.currentTimeMillis();

			if (buffer.size() > getBufferSize()
					|| now - lastFlushTime >= getFlushInternal()) {
				synchronized (buffer) {
					try {
						flush(buffer);
					} catch (IOException e) {
						LOGGER.error("flush buffer error: " + e.toString());
						
						buffer.clear();
					} finally {
						lastFlushTime = now;

						buffer.notifyAll();
					}
				}
			}

			try {
				Thread.sleep(getSleepTime());
			} catch (InterruptedException e) {
			}
		}

		synchronized (buffer) {
			try {
				flush(buffer);
			} catch (IOException e) {
				LOGGER.error("flush buffer error: " + e.toString());
			}
		}

		try {
			onStop();
		} catch (Exception e) {
			throw new RuntimeException(Statics.HOSTNAME + "\t"
					+ Thread.currentThread().getName() + "\t" + e.toString());
		}
	}

	public void requestStop() {
		this.stop = true;
	}

}
